package com.yanggu.flink.datastream_api.multi_stream_transform.split_stream

import com.yanggu.flink.datastream_api.source.ClickSource
import org.apache.flink.streaming.api.scala._

/**
 * 通过filter算子实现分流操作
 * 本质上是将流数据复制成了多份
 */
object SplitStreamByFilterDemo {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val clickSource = env.addSource(new ClickSource)

    val maryStream = clickSource.filter(event => event.name == "Mary")
    val bobStream = clickSource.filter(event => event.name == "Bob")
    val otherStream = clickSource.filter(event => event.name != "Mary" && event.name != "Bob")

    maryStream.print("Mary pv")
    bobStream.print("Bob pv")
    otherStream.print("other pv")

    env.execute()

  }

}
